package com.douyin.service.impl;

import com.douyin.constant.RedisConstant;
import com.douyin.entity.User;
import com.douyin.entity.video.Video;
import com.douyin.service.InterestPushService;
import com.douyin.service.TypeService;
import com.douyin.utils.RedisCacheUtil;
import com.douyin.vo.Model;
import com.douyin.vo.UserModel;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisCallback;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.util.ObjectUtils;

import java.util.*;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

@Service
public class InterestPushServiceImpl implements InterestPushService {

    @Autowired
    private RedisCacheUtil redisCacheUtil;
    @Autowired
    private RedisTemplate redisTemplate;
    @Autowired
    private TypeService typeService;
    @Override
    @Async
    public void pushSystemStockIn(Video video) {
        // 往系统库中，key=添加视频对应的每个标签，value=视频id
        List<String> list = video.buildLabel();
        Long videoId = video.getId();
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            for (String label : list) {
                connection.sAdd((RedisConstant.SYSTEM_STOCK + label).getBytes(),String.valueOf(videoId).getBytes());
            }
            return null;
        });
    }

    //分片解决Redis大key问题
    @Override
    @Async
    public void pushSystemTypeStockIn(Video video) {
        final Long typeId = video.getTypeId();
        final String currentIdKey = RedisConstant.SYSTEM_TYPE_STOCK_ID_PREFIX + typeId;
        Long currentId = getOrCreateCurrentId(currentIdKey);
        final String key = RedisConstant.SYSTEM_TYPE_STOCK + typeId + ":" + currentId;
        redisCacheUtil.sSet(key, video.getId());

        // 检查是否需要创建新的分片
        Long size = redisTemplate.opsForSet().size(key);
        if (size >= 5000) {
            incrementCurrentId(currentIdKey);
        }
    }

    //从较新的分片获得视频
    @Override
    public Collection<Long> listVideoIdByTypeId(Long typeId) {
        final String currentIdKey = RedisConstant.SYSTEM_TYPE_STOCK_ID_PREFIX + typeId;
        Long currentId = getOrCreateCurrentId(currentIdKey);
        List<Long> newList =null;
        List<Long> oldList = new ArrayList<>();
        if (currentId==1){
            newList =getRandomVideoIds(RedisConstant.SYSTEM_TYPE_STOCK + typeId+ ":" + currentId, 10);
        }else {
            newList=getRandomVideoIds(RedisConstant.SYSTEM_TYPE_STOCK + typeId+ ":" + currentId, 7);
            Random random = new Random();
            for (int i = 0; i < 3; i++) {
                int index = random.nextInt(newList.size());
                oldList.add(Long.parseLong(redisTemplate.opsForSet().randomMembers(
                        RedisConstant.SYSTEM_TYPE_STOCK + typeId+ ":" + currentId,1).toString()));
            }
        }
        // 合并结果并去重
        Set<Long> result = new HashSet<>();
        result.addAll(newList);
        result.addAll(oldList);
        return result;
    }


    @Override
    @Async
    public void deleteSystemStockIn(Video video) {
        final List<String> labels = video.buildLabel();
        final Long videoId = video.getId();
        redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
            for (String label : labels) {
                connection.sRem((RedisConstant.SYSTEM_STOCK + label).getBytes(),String.valueOf(videoId).getBytes());
            }
            return null;
        });
    }

    @Override
    public Collection<Long> listVideoIdByLabels(List<String> labelNames) {
        ArrayList<String> labelKeys = new ArrayList<>();
        for (String labelName : labelNames) {
            labelKeys.add(RedisConstant.SYSTEM_STOCK + labelName);
        }
        Set<Long> videoIds = new HashSet<>();
        List<Object> list = redisCacheUtil.sRandom(labelKeys);
        if (!ObjectUtils.isEmpty(list)){
            videoIds = list.stream().filter(id ->!ObjectUtils.isEmpty(id)).map(id -> Long.valueOf(id.toString())).collect(Collectors.toSet());
        }
        return videoIds;
    }

    @Override
    @Async
    public void deleteSystemTypeStockIn(Video video) {
        final Long typeId = video.getTypeId();
        final String currentIdKey = RedisConstant.SYSTEM_TYPE_STOCK_ID_PREFIX + typeId;
        Long currentId = getOrCreateCurrentId(currentIdKey);
        for (int i = 1; i <= currentId; i++){
            if (redisTemplate.opsForSet().isMember(RedisConstant.SYSTEM_TYPE_STOCK + typeId + ":" + i, video.getId())){
                redisCacheUtil.setRemove(RedisConstant.SYSTEM_TYPE_STOCK + typeId + ":" + i, video.getId());
                break;
            }
        }
    }

    /**
     * 初始化用户模型
     * 根据用户ID和标签列表来初始化用户模型，在Redis中存储每个标签对应的初始概率值
     *
     * @param userId 用户ID，用于在Redis中存储用户模型时生成唯一的key
     * @param labels 用户的标签列表，用于初始化用户模型
     */
    @Override
    public void initUserModel(Long userId, List<String> labels) {
        //使用redis的原因是用户模型需要频繁更新，所以使用redis存储
        String key = RedisConstant.USER_MODEL + userId;
        Map<Object, Object> modelMap = new HashMap<>();
        if (!ObjectUtils.isEmpty(labels)) {
            final int size = labels.size();
            // 将标签分为等分概率,不可能超过100个分类
            double probabilityValue = 100 / size;
            for (String labelName : labels) {
                modelMap.put(labelName, probabilityValue);
            }
        }
        redisCacheUtil.del(key);
        redisCacheUtil.hmset(key, modelMap);
        // 为用户模型设置ttl
    }

    @Override
    public void updateUserModel(UserModel userModel) {
        Long userId = userModel.getUserId();
        //游客不需要管
        if (userId != null){
            final List<Model> models = userModel.getModels();
            String key = RedisConstant.USER_MODEL + userId;
            Map<Object, Object> modelMap = redisCacheUtil.hmget(key);
            if (modelMap==null){
                modelMap = new HashMap<>();
            }
            for (Model model : models) {
                if (modelMap.containsKey(model.getLabel())){
                    modelMap.put(model.getLabel(),Double.parseDouble(modelMap.get(model.getLabel()).toString())+model.getScore());
                    Object object = modelMap.get(model.getLabel());
                    //分数可能为负数，所以需要判断
                    if (object==null||Double.parseDouble(object.toString())<=0){
                        //移除标签
                        modelMap.remove(model.getLabel());
                    }
                }else {
                    modelMap.put(model.getLabel(),model.getScore());
                }
            }

            // 每个标签概率同等加上标签数，再同等除以标签数  防止数据膨胀
            int labelSize = modelMap.keySet().size();
            for (Object o : modelMap.keySet()) {
                modelMap.put(o,(Double.parseDouble(modelMap.get(o).toString()) + labelSize )/ labelSize);
            }
            // 更新用户模型
            redisCacheUtil.hmset(key, modelMap);
        }
    }

    @Override
    public Collection<Long> listVideoIdByUserModel(User user) {
        // 创建结果集
        Set<Long> videoIds = new HashSet<>(10);

        if (user != null) {
            final Long userId = user.getId();
            // 从模型中拿概率
            final Map<Object, Object> modelMap = redisCacheUtil.hmget(RedisConstant.USER_MODEL + userId);
            if (!ObjectUtils.isEmpty(modelMap)) {
                // 组成数组
                final String[] probabilityArray = initProbabilityArray(modelMap);
                final Boolean sex = user.getSex();
                // 获取视频
                final Random randomObject = new Random();
                final ArrayList<String> labelNames = new ArrayList<>();
                // 随机获取X个视频
                for (int i = 0; i < 8; i++) {
                    String labelName = probabilityArray[randomObject.nextInt(probabilityArray.length)];
                    labelNames.add(labelName);
                }
                // 提升性能
                String t = RedisConstant.SYSTEM_STOCK;
                // 随机获取
                //executePipelined方法批量执行Redis命令，提升性能。
                List<Object> list = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                    for (String labelName : labelNames) {
                        String key = t + labelName;
                        //调用connection.sRandMember从集合中随机获取一个成员。
                        connection.sRandMember(key.getBytes());
                    }
                    return null;
                });
                // 获取到的videoIds
                Set<Long> ids = list.stream().filter(id->id!=null).map(id->Long.parseLong(id.toString())).collect(Collectors.toSet());
                String key2 = RedisConstant.HISTORY_VIDEO;

                // 查看已经看过的视频，去重
                List simpIds = redisTemplate.executePipelined((RedisCallback<Object>) connection -> {
                    for (Long id : ids) {
                        connection.get((key2 + id + ":" + userId).getBytes());
                    }
                    return null;
                });
                simpIds = (List) simpIds.stream().filter(o->!ObjectUtils.isEmpty(o)).collect(Collectors.toList());;
                if (!ObjectUtils.isEmpty(simpIds)){
                    for (Object simpId : simpIds) {
                        final Long l = Long.valueOf(simpId.toString());
                        if (ids.contains(l)){
                            ids.remove(l);
                        }
                    }
                }
                //TODO 可以选择布隆过滤器去重

                videoIds.addAll(ids);

                // 随机挑选一个视频,根据性别: 男：美女 女：宠物
                final Long aLong = randomVideoId(sex);
                if (aLong!=null){
                    videoIds.add(aLong);
                }


                return videoIds;
            }
        }
        // 游客
        // 随机获取10个标签
        final List<String> labels = typeService.random10Labels();
        final ArrayList<String> labelNames = new ArrayList<>();
        int size = labels.size();
        final Random random = new Random();
        // 获取随机的标签
        for (int i = 0; i < 10; i++) {
            final int randomIndex = random.nextInt(size);
            labelNames.add(RedisConstant.SYSTEM_STOCK + labels.get(randomIndex));
        }
        // 获取videoId
        final List<Object> list = redisCacheUtil.sRandom(labelNames);
        if (!ObjectUtils.isEmpty(list)){
            videoIds = list.stream().filter(id ->!ObjectUtils.isEmpty(id)).map(id -> Long.valueOf(id.toString())).collect(Collectors.toSet());
        }

        return videoIds;
    }




    // 初始化概率数组 -> 保存的元素是标签
    public String[] initProbabilityArray(Map<Object, Object> modelMap) {
        // key: 标签  value：概率
        Map<String, Integer> probabilityMap = new HashMap<>();
        int size = modelMap.size();
        final AtomicInteger n = new AtomicInteger(0);
        modelMap.forEach((k, v) -> {
            // 防止结果为0,每个同等加上标签数
            int probability = (((Double) v).intValue() + size) / size;
            n.getAndAdd(probability);
            probabilityMap.put(k.toString(), probability);
        });
        final String[] probabilityArray = new String[n.get()];

        final AtomicInteger index = new AtomicInteger(0);
        // 初始化数组,初始化后数组大概为[1,1,1,2,2,3,3,3,3]
        probabilityMap.forEach((labelsId, p) -> {
            int i = index.get();
            int limit = i + p;
            while (i < limit) {
                probabilityArray[i++] = labelsId;
            }
            index.set(limit);
        });
        return probabilityArray;
    }

    public Long randomVideoId(Boolean sex) {
        String key = RedisConstant.SYSTEM_STOCK + (sex ? "美女" : "宠物");
        final Object o = redisCacheUtil.sRandom(key);
        if (o!=null){
            return Long.parseLong(o.toString());
        }
        return null;
    }

    private Long getOrCreateCurrentId(String currentIdKey) {
        Object o = redisTemplate.opsForValue().get(currentIdKey);
        Long currentId =null;
        if (o == null){
            currentId = 1L;
            redisTemplate.opsForValue().set(currentIdKey, String.valueOf(currentId));
            return currentId;
        }
        currentId = Long.parseLong(o.toString());
        return currentId;
    }

    private void incrementCurrentId(String currentIdKey) {
        redisTemplate.opsForValue().increment(currentIdKey);
    }

    private List<Long> getRandomVideoIds(String key, int count) {
        final List<Object> list = redisTemplate.opsForSet().randomMembers(key, count);
        List<Long> result = new ArrayList<>();
        for (Object aLong : list) {
            if (aLong != null) {
                result.add(Long.parseLong(aLong.toString()));
            }
        }
        return result;
    }
}
